package kubernetes

import (
	"errors"
	"fmt"
	"github.com/sirupsen/logrus"
	"github.com/tluo-github/super-runner/common"
	"github.com/tluo-github/super-runner/executors"
	k8s_helper "github.com/tluo-github/super-runner/helpers/k8s"
	"github.com/tluo-github/super-runner/helpers/kafka"
	"github.com/tluo-github/super-runner/helpers/minio"
	api "k8s.io/api/core/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/client-go/kubernetes"
	"os"
	"path/filepath"
	"time"
)

type kubernetesOptions struct {
	Image    common.Image    //build image
	Services common.Services // services image
}

type KubernetesExecutor struct {
	Config      common.RunnerConfig
	Build       common.Build
	kubeClient  *kubernetes.Clientset
	pod         *api.Pod
	pod_name    string
	options     *kubernetesOptions
	command     string
	buildFinish chan error
	BuildLog    *os.File
	ObjectName  string
	IsSystemError bool
}

func (s *KubernetesExecutor) Prepare(config common.RunnerConfig, build common.Build) error {
	s.Config = config
	s.Build = build
	s.buildFinish = make(chan error, 1)
	s.IsSystemError = false;

	s.pod_name = fmt.Sprintf("super-runner-%s-%d",s.Build.Name,s.Build.Timestamp)
	// 获得要执行的命令
	s.command = build.Command

	// 创建 build log
	filename := fmt.Sprintf("/logs/%s.log",s.pod_name)
	build_log, err := os.Create(filename)

	if err != nil {
		return err
	}
	s.BuildLog = build_log

	timeStr:=time.Now().Format("2006-01-02")
	s.ObjectName = fmt.Sprintf("%s/%s",timeStr,filepath.Base(s.BuildLog.Name()))

	// 发送开始执行消息给 kafka
	kafka.StartKafka(config.KafkaConfig,s.pod_name,build.MetaData,config.Name)

	// 连接 K8s cluster
	s.kubeClient, err = k8s_helper.GetKubeClient(s.Config.Kubernetes)
	if err != nil{
		logrus.WithFields(logrus.Fields{
			"pod_name": s.pod_name,
		}).Errorln("connection k8s faild with error:", err)
		s.IsSystemError = true
		return errors.New("connection k8s faild with error")
	}
	// 处理 options config 相关
	s.prepareOptions(build)

	return nil
}

func (s *KubernetesExecutor) prepareOptions(build common.Build)  {
	s.options = &kubernetesOptions{}
	s.options.Image = build.Image
	for _, service := range build.Services {
		if service.Name == "" {
			continue
		}
		s.options.Services = append(s.options.Services, service)
	}
}

func (s *KubernetesExecutor) Run() error {
	// k8s pod 初始化
	err := s.setupBuildPod()
	if err != nil {
		logrus.WithFields(logrus.Fields{
			"pod_name": s.pod_name,
		}).Errorln("setupBuildPod with error: ", err)
		s.IsSystemError = true
		return err
	}


	// 开始通过 k8s client 让 pod 执行命令
	s.runInContainer()
	return nil
}
func Test()  {
	config := &common.Kubernetes{
		Config:    "/Users/luotao/.kube/config",
		Namespace: "",
	}
	kubeConfig,err := k8s_helper.GetKubeClient(*config)
	if err != nil{
		panic(err)
	}
	pod, error :=kubeConfig.CoreV1().Pods("default").Get("nginx-dm-5cd679bc7b-xxx",metav1.GetOptions{})
	if error != nil{
		panic(error)
	}
	logrus.Infoln(pod)
}
// 初始化设置 pod
func (s *KubernetesExecutor) setupBuildPod() error {

	logrus.WithFields(logrus.Fields{
		"pod_name": s.pod_name,
	}).Info("setupBuildPod")

	services := make([]api.Container, len(s.options.Services))
	for i, service := range s.options.Services {
		services[i] = s.buildContainer(fmt.Sprintf("svc-%d", i), service.Name)
	}
	//todo step labels
	//todo step annotations
	//todo step imagePullSecrets
	command := []string{"sh","-c","tail -f /dev/null"}
	pod_container := s.buildContainer("build", s.options.Image.Name)
	pod_container.Command = command

	PodDNSConfig := api.PodDNSConfig{
		Nameservers: s.Config.Kubernetes.DNS,
		Searches:    nil,
		Options:     nil,
	}

	pod_resource := &api.Pod{
		ObjectMeta : metav1.ObjectMeta{
			Name:s.pod_name,
			Namespace: s.Config.Kubernetes.Namespace,
		},
		Spec: api.PodSpec{
			Volumes: s.getVolumes(),
			Containers: append([]api.Container{
				pod_container,
			}, services...),
			RestartPolicy: api.RestartPolicyNever,
			DNSConfig:  &PodDNSConfig,
		},
	}
	// 创建 pod,添加重试功能
	pod, err := s.kubeClient.CoreV1().Pods(s.Config.Kubernetes.Namespace).Create(pod_resource)
	if err != nil {
		logrus.WithFields(logrus.Fields{
			"pod_name": s.pod_name,
		}).Warnln("setupBuildPod create pod [1] with error: ", err)

		time.Sleep(30 * time.Second)
		pod, err =s.kubeClient.CoreV1().Pods(s.Config.Kubernetes.Namespace).Get(s.pod_name,metav1.GetOptions{})
		if err != nil {
			pod, err = s.kubeClient.CoreV1().Pods(s.Config.Kubernetes.Namespace).Create(pod_resource)
			if err != nil{
				logrus.WithFields(logrus.Fields{
					"pod_name": s.pod_name,
				}).Warnln("setupBuildPod create pod [2] with error: ", err)

				time.Sleep(30 * time.Second)
				pod, err =s.kubeClient.CoreV1().Pods(s.Config.Kubernetes.Namespace).Get(s.pod_name,metav1.GetOptions{})
				if err != nil {
					pod, err = s.kubeClient.CoreV1().Pods(s.Config.Kubernetes.Namespace).Create(pod_resource)
					if err != nil {
						return err
					}
				}

			}
		}

	}
	s.pod = pod
	return nil

}
// 生成容器
func (s *KubernetesExecutor) buildContainer(
	name string,
	image string, )  api.Container {
	privileged := true


	liveness := api.Probe{
		Handler:  api.Handler{
			Exec: &api.ExecAction{Command: []string{"sh","-c","kill me"}},
		},
		InitialDelaySeconds: s.getTimeout(),
		TimeoutSeconds:      1,
		PeriodSeconds:       1,
		SuccessThreshold:    1,
		FailureThreshold:    1,
	}
	return api.Container{
		Name:                     name,
		Image:                    image,
		Resources:                api.ResourceRequirements{},
		ImagePullPolicy:          api.PullIfNotPresent,
		SecurityContext:          &api.SecurityContext{
			Privileged: &privileged ,
		},
		Env:             k8s_helper.BuildVariables(s.Build.Variables),
		VolumeMounts:			  s.getVolumeMounts(),
		Stdin:                    true,
		LivenessProbe: &liveness,

	}
}
// 获得所有 VolumeMounts
func (s *KubernetesExecutor) getVolumeMounts() (mounts []api.VolumeMount) {
	for _, mount := range s.Build.Kubernetes.Volumes.Host_paths {
		mounts = append(mounts, api.VolumeMount{
			Name:      mount.Name,
			MountPath: mount.Mount_path,
			ReadOnly:  mount.Read_only,
		})
	}
	return
}
// 获得所有的 Volume
func (s *KubernetesExecutor) getVolumes() (volumes []api.Volume) {
	for _, volume := range s.Build.Kubernetes.Volumes.Host_paths {
		path := volume.Host_path
		// Make backward compatible with syntax introduced in version 9.3.0
		if path == "" {
			path = volume.Host_path
		}

		volumes = append(volumes, api.Volume{
			Name: volume.Name,
			VolumeSource: api.VolumeSource{
				HostPath: &api.HostPathVolumeSource{
					Path: path,
				},
			},
		})
	}
	return
}


func (s *KubernetesExecutor) runInContainer()   {
	go func() {
		status, err := k8s_helper.WaitForPodRunning(s.kubeClient, s.pod)
		if err != nil {
			s.buildFinish <- err
			return
		}
		if status != api.PodRunning {
			s.IsSystemError = true
			s.buildFinish <- fmt.Errorf("pod failed to enter running state: %s", status)
			return
		}

		logrus.WithFields(logrus.Fields{
			"pod_name": s.pod_name,
		}).Info("pod state is PodRunning")

		config, err := k8s_helper.GetKubeClientConfig(s.Config.Kubernetes)
		if err != nil {
			s.buildFinish <- err
			return
		}
		cmd := []string{
			"sh",
			"-c",
			//"set -e pipefail ; pwd",
			s.command,
		}
		exec := ExecOptions{
			PodName:       s.pod.Name,
			Namespace:     s.pod.Namespace,
			ContainerName: "build",
			Command:       cmd,
			In:            nil,
			Out:           s.BuildLog,
			Err:           s.BuildLog,
			Stdin:         true,
			Config:        config,
			Client:        s.kubeClient,
			Executor:      &DefaultRemoteExecutor{},
		}
		s.buildFinish <- exec.Run()
	}()
}

func (s *KubernetesExecutor) Wait() error {
	// Wait for signals: abort, timeout or finish
	select {
	case <- time.After( time.Duration(s.getTimeout()) * time.Second ):
		logrus.WithFields(logrus.Fields{
			"pod_name": s.pod_name,
		}).Warn("Build Timeout")
		url := fmt.Sprintf("%s/minio/%s/%s",s.Config.MinioConfig.Url,s.Config.MinioConfig.Bucket,s.ObjectName)
		kafka.ErrorKafka(s.Config.KafkaConfig,s.pod_name,s.Build.MetaData,url,errors.New("timeout"),s.IsSystemError)

	case err := <- s.buildFinish:
		// command finished
		if err != nil{
			logrus.WithFields(logrus.Fields{
				"pod_name": s.pod_name,
			}).Warn("build faild with error: ", err)
			url := fmt.Sprintf("%s/minio/%s/%s",s.Config.MinioConfig.Url,s.Config.MinioConfig.Bucket,s.ObjectName)
			kafka.ErrorKafka(s.Config.KafkaConfig,s.pod_name,s.Build.MetaData,url,err,s.IsSystemError)

		} else {
			logrus.WithFields(logrus.Fields{
				"pod_name": s.pod_name,
			}).Info("Build succeeded")
			url := fmt.Sprintf("%s/minio/%s/%s",s.Config.MinioConfig.Url,s.Config.MinioConfig.Bucket,s.ObjectName)
			kafka.SucceededKafka(s.Config.KafkaConfig,s.pod_name,s.Build.MetaData,url)
		}
	}
	return nil
}

func (s* KubernetesExecutor)SendError(err error)  {
	logrus.WithFields(logrus.Fields{
		"name": s.pod_name,
		"error": err,
	}).Infoln("SendError")

	url := fmt.Sprintf("%s/minio/%s/%s",s.Config.MinioConfig.Url,s.Config.MinioConfig.Bucket,s.ObjectName)
	kafka.ErrorKafka(s.Config.KafkaConfig,s.pod_name,s.Build.MetaData,url,err,s.IsSystemError)
	s.BuildLog.WriteString(err.Error())
}
func (s *KubernetesExecutor) cleanBuildLog()  {
	if s.BuildLog != nil {
		if minio.NewMinioHelper(s.Config.MinioConfig) != nil{
			_,err := minio.NewMinioHelper(s.Config.MinioConfig).PutObject(s.ObjectName,s.BuildLog.Name())
			if err != nil {
				logrus.WithFields(logrus.Fields{
					"pod_name": s.pod_name,
				}).Error("Minio PutObject with error:", err)
			}
			logrus.WithFields(logrus.Fields{
				"pod_name": s.pod_name,
				"objectName": s.ObjectName,
			}).Infoln("Minio PutObject ok")
		}
		os.Remove(s.BuildLog.Name())
	}
}
func (s *KubernetesExecutor) Cleanup() error {
	if s.pod != nil{
		err := s.kubeClient.CoreV1().Pods(s.pod.Namespace).Delete(s.pod.Name, &metav1.DeleteOptions{})
		if err != nil {
			time.Sleep(5 * time.Second)
			err = s.kubeClient.CoreV1().Pods(s.pod.Namespace).Delete(s.pod.Name, &metav1.DeleteOptions{})
			if err != nil {
				time.Sleep(5 * time.Second)
				err = s.kubeClient.CoreV1().Pods(s.pod.Namespace).Delete(s.pod.Name, &metav1.DeleteOptions{})
				if err != nil {
					logrus.WithFields(logrus.Fields{
						"pod_name": s.pod.Name,
					}).Warn("Error cleaning up pod with error: ", err)
				}
			}

		}
	}
	k8s_helper.CloseKubeClient(s.kubeClient)

	logrus.WithFields(logrus.Fields{
		"pod_name": s.pod_name,
	}).Info("Pod Cleanup")


	s.cleanBuildLog()
	return nil
}

func (s *KubernetesExecutor) getTimeout ()int32 {
	if s.Build.Timeout > 0{
		return s.Build.Timeout
	}
	return common.DefaultLivessTimeout

}
func createFn() common.Executor {
	return &KubernetesExecutor{}
}
func init()  {
	common.RegisterExecutor("kubernetes",
		executors.DefaultExecutorProvider{Creator: createFn})
}